package com.headfirst.dmp.report.AreanalyzeRpt

import com.headfirst.dmp.beans.LogBean
import com.headfirst.dmp.utils.AreaAnalyzeUtils
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 分析统计按照不同地域的各项指标
  *
  * 使用工具：Spark-Core
  *
  * 与V2版不同的是，本次读取的是原始文件，不是parquet文件
  */
object AreaAnalyzeRPT_V3 {

  def main(args: Array[String]): Unit = {

    //1.验证参数
    if (args.length != 2) {
      print(
        """
          |com.headfirst.dmp.report.AreaAnalyzeRPT_V3
          |需要参数：
          |       logInputPath
          |       resultOutputPath
        """.stripMargin)
      sys.exit(-1)
    }

    //2.接受参数
    val Array(logInputPath, resultOutputPath) = args

    //3.创建sparkcontext对象
    val conf = new SparkConf()
    conf.setMaster("local[*]") //本地测试使用local，提交到集群则注释掉该配置
    conf.setAppName("AreaAnalyzeRPT")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //设置序列化方式采用KryoSerializer方式（默认的是java序列化）

    val sc = new SparkContext(conf)

    val data: RDD[((String, String), List[Double])] = sc.textFile(logInputPath)
      .map(x => x.split(",", -1))
      .filter(x => x.length >= 85)
      .map(arr => {
        val log = LogBean(arr)

        //原始请求、有效请求、广告请求
        val reqList: List[Double] = AreaAnalyzeUtils.caculateReq(log.requestmode, log.processnode)
        //参与竞价的次数，参与竞价的次数
        val bidList: List[Double] = AreaAnalyzeUtils.caculateBid(log.adplatformproviderid, log.iseffective, log.isbilling, log.isbid, log.adorderid, log.iswin)
        //展示数,点击数
        val showList: List[Double] = AreaAnalyzeUtils.caculateShowAndClick(log.requestmode, log.iseffective)
        //广告消费 ，广告成本
        val costList: List[Double] = AreaAnalyzeUtils.caculateCost(log.adplatformproviderid, log.iseffective, log.isbilling, log.iswin, log.adorderid, log.adcreativeid, log.winprice, log.adpayment)

        //返回值
        ((log.provincename, log.cityname), reqList ++ bidList ++ showList ++ costList)
      })


    //将数据进行分组聚合，这里使用的是zip操作
    //zip两个list，list1=1,2,3   list2=2,4,6  得到：list3((1,2),(3,4),(5,6))
    data.reduceByKey((list1, list2) => {
      list1.zip(list2).map(t => t._1 + t._2)
    }).map(t => {
      t._1._1 + "," + t._1._2 + "," + t._2.mkString(",")
    }).saveAsTextFile(resultOutputPath)


    //关流
    sc.stop()

  }

}
